package com.example.micro.thread.queue.pool;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * 消费者线程
 *
 * @author Code Farmer
 * @create 2021/3/29 1:21 上午
 */
public class ThreadCustomer implements Callable<Integer> {

    /**
     * 数据队列
     */
    ArrayBlockingQueue<Integer> dataQueue;

    public ThreadCustomer(ArrayBlockingQueue<Integer> dataQueue) {
        this.dataQueue = dataQueue;
    }

    @Override
    public Integer call() throws Exception {
        //循环消费处理数据
        while (true) {
            //从数据队列中取数据
            int data = dataQueue.take();
            System.out.println("消费线程-" + Thread.currentThread().getName() + "获取到待处理数据:" + data);
            try {
                int result = dispiseData(data);
                System.out.println("消费线程-" + Thread.currentThread().getName() + "处理数据" + data + "的结果为:" + result);  //正常处理结果
            } catch (Exception e) {
                System.out.println("消费线程-" + Thread.currentThread().getName() + "处理数据" + data + "出错,原因为:" + e); //异常处理
            }
        }
    }

    /**
     * 处理数据（从数据队列中获取得到的）
     *
     * @param data
     * @return
     * @throws Exception
     */
    Integer dispiseData(Integer data) throws Exception {
        ExecutorService threadPool = Executors.newFixedThreadPool(5);  //生成处理线程线池（5个线程）
        List<Callable<Integer>> cdCallList = new ArrayList<>();  //处理线程 5个
        for (int i = 0; i < 5; i++) {
            cdCallList.add(new ThreadCustomerDispose(data));
        }

        //InvokeAll方法 批量处理任务，获取5个Future结果
        List<Future<Integer>> resultFutures = threadPool.invokeAll(cdCallList);
        int result = 0;
        //计算处理结果
        for (Future<Integer> t : resultFutures) {
            result += t.get();
        }

        /**
         * 如果调用了shutdown()方法，则线程池处于SHUTDOWN状态，此时线程池不能够接受新的任务，它会等待所有任务执行完毕；
         *
         * 如果调用了shutdownNow()方法，则线程池处于STOP状态，此时线程池不能接受新的任务，并且会去尝试终止正在执行的任务；
         */
        threadPool.shutdown();
        return result;
    }
}
